package com.atguigu.fail.detect;

import com.atguigu.fail.detect.bean.LoginEvent;
import com.atguigu.fail.detect.bean.LoginFailWarning;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.curator.org.apache.curator.shaded.com.google.common.collect.Lists;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.net.URL;
import java.util.ArrayList;
import java.util.Iterator;

public class LoginFail {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        //1.从文件中读取数据
        URL resource = LoginFail.class.getResource("/LoginLog.csv");
        DataStreamSource<String> dataStreamSource = env.readTextFile(resource.getPath());
        SingleOutputStreamOperator<LoginEvent> loginEventStream = dataStreamSource.map(line -> {
            String[] fields = line.split(",");
            return new LoginEvent(new Long(fields[0]), fields[1], fields[2], new Long(fields[3]));
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<LoginEvent>(Time.seconds(3)) {
            @Override
            public long extractTimestamp(LoginEvent element) {
                return element.getTimestamp() * 1000L;
            }
        });

        //自定义处理函数，检测连续登陆失败事件
        SingleOutputStreamOperator<LoginFailWarning> warningStream = loginEventStream
                .keyBy(LoginEvent::getUserId)
                .process(new LoginFailDetectWarning(2));

        warningStream.print();
        env.execute("login event warning");
    }

    public static class LoginFailDetectWarning0 extends KeyedProcessFunction<Long, LoginEvent, LoginFailWarning> {
        //定义连续登陆失败次数
        private Integer maxFailTimes;

        public LoginFailDetectWarning0(Integer maxFailTimes) {
            this.maxFailTimes = maxFailTimes;
        }

        //定义状态，保存当前2秒内所有的登陆失败事件
        ListState<LoginEvent> loginFailEventListState;
        //定义状态，保存注册的定时器时间戳;
        ValueState<Long> timerTsState;

        @Override
        public void open(Configuration parameters) throws Exception {
            loginFailEventListState = getRuntimeContext().getListState(new ListStateDescriptor<LoginEvent>("login-fail-event", LoginEvent.class));
            timerTsState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timer-ts", Long.class));
        }

        @Override
        public void processElement(LoginEvent value, Context ctx, Collector<LoginFailWarning> out) throws Exception {
            //判断登陆事件的类型
            if ("fail".equals(value.getLoginState())) {
                //如果失败，添加到列表状态中
                loginFailEventListState.add(value);
                //如果没有定时器注册一个2秒的定时器
                if (timerTsState.value() == null) {
                    Long ts = value.getTimestamp() + 2 * 1000L;
                    ctx.timerService().registerEventTimeTimer(ts);
                    timerTsState.update(ts);
                }
            } else {
                //删除定时器，清空状态
                if (timerTsState.value() != null) {
                    ctx.timerService().deleteEventTimeTimer(timerTsState.value());
                }
                loginFailEventListState.clear();
                timerTsState.clear();
            }
        }


        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<LoginFailWarning> out) throws Exception {
            //定时触发，说明2秒内没有登陆成功，判断list state中失败个数
            ArrayList<LoginEvent> loginFailEvents = Lists.newArrayList(loginFailEventListState.get());

            Integer failTimes = loginFailEvents.size();

            //如果超出设定的最大失败次数，输出报警
            if (failTimes >= maxFailTimes) {
                out.collect(new LoginFailWarning(ctx.getCurrentKey()
                        , loginFailEvents.get(0).getTimestamp()
                        , loginFailEvents.get(failTimes - 1).getTimestamp()
                        , "login failed " + failTimes + " times in 2 seconds"));
            }

            //清空状态
            loginFailEventListState.clear();
            timerTsState.clear();
        }
    }

    public static class LoginFailDetectWarning extends KeyedProcessFunction<Long, LoginEvent, LoginFailWarning> {

        //定义连续登陆失败次数
        private Integer maxFailTimes;

        public LoginFailDetectWarning(Integer maxFailTimes) {
            this.maxFailTimes = maxFailTimes;
        }

        //定义状态，保存当前2秒内所有的登陆失败事件
        ListState<LoginEvent> loginFailEventListState;

        @Override
        public void open(Configuration parameters) throws Exception {
            loginFailEventListState = getRuntimeContext().getListState(new ListStateDescriptor<LoginEvent>("login-fail-event", LoginEvent.class));
        }

        @Override
        public void processElement(LoginEvent value, Context ctx, Collector<LoginFailWarning> out) throws Exception {
            //以登陆事件作为判断报警的登录触发时间，不再注册定时器
            //判断当前时间登录状态
            if ("fail".equals(value.getLoginState())) {
                //1.如果登录失败，获取状态中之前的登陆失败时间，继续判断是否已有失败事件
                Iterator<LoginEvent> iterator = loginFailEventListState.get().iterator();
                if (iterator.hasNext()) {
                    //1.1如果已经有登录失败判断时间戳是否在2秒内
                    // 获取已有的登陆失败事件
                    LoginEvent firstFailEvent = iterator.next();
                    if (value.getTimestamp() - firstFailEvent.getTimestamp() <= 2) {
                        out.collect(new LoginFailWarning(value.getUserId()
                                , firstFailEvent.getTimestamp()
                                , value.getTimestamp()
                                , "login fail 2 times in 2 seconds"));
                    }
                    //不论是否报警，都需要更新状态
                    loginFailEventListState.clear();
                    loginFailEventListState.add(value);
                } else {
                    //1.2如果没有登陆失败，直接将当前事件存入list state
                    loginFailEventListState.add(value);
                }

            } else {
                //2.登录成功就清空状态
                loginFailEventListState.clear();
            }
        }
    }
}

